package org.keystate

import org.FlinkStreamApp
import org.apache.flink.streaming.api.scala._
import org.bean.SensorReading
import org.diysource.SensorSource

/**
 * description ：key state2 example类
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/24 20:24
 * modified By ：
 * version:    : 1.0
 */
object KeyedStateExample2 extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 获取源数据
    val source = environment.addSource(new SensorSource)

    // 操作
    source
      .keyBy(_.id)
      .flatMapWithState[(String, Double, Double), Double] {
        case (in: SensorReading, None) => (List.empty, Some(in.temperature))
        case (in: SensorReading, lastTemp: Some[Double]) =>
          val diff = (lastTemp.get - in.temperature).abs
          if (diff > 2) {
            (List((in.id, lastTemp.get, in.temperature)), Some(in.temperature))
          } else {
            (List.empty, Some(in.temperature))
          }
      }
      .print
  }
}
